package com.haishuibutian.rocketmqproducer.rocketmq;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.UUID;


@Component
public class MQProducer<T> {

    private static final Logger logger = LoggerFactory.getLogger(MQProducer.class);

    private DefaultMQProducer producer;

    @Resource
    private MQConfig mqConfig;

    @PostConstruct
    public void init() throws MQClientException {
        try {
            logger.info("MQ生产者初始化...");
            producer = new DefaultMQProducer(mqConfig.getGroupName());
            producer.setNamesrvAddr(mqConfig.getNameSrv());
            producer.setRetryTimesWhenSendFailed(mqConfig.getMaxReConsumeTimes());
            producer.start();
            logger.info("MQ生产者启动成功 -> nameSrv={},groupName={}", mqConfig.getNameSrv(), mqConfig.getGroupName());
        } catch (MQClientException e) {
            logger.error("MQ生产者启动失败", e);
            throw e;
        }
    }

    @PreDestroy
    public void destroy() {
        if (producer != null) {
            producer.shutdown();
            logger.info("MQ生产者销毁...");
        }
    }

    public void sendMessage(T body, String topic, String tag) {
        try {
            logger.info("发送MQ消息,topic:{},tag:{},body:{}", topic, tag, body);
            send(body, topic, tag);
        } catch (FailureInSendingToMQException e) {
            logger.error("MQ消息发送失败，body={} ", body, e);
            throw new MQException("MQ消息发送失败", e);
        }
    }

    public <V> void sendDelay(V obj, String topic, String tag) throws FailureInSendingToMQException {
        send(JSON.toJSONString(obj).getBytes(), topic, tag, true);
    }

    public <V> void send(V obj, String topic, String tag) throws FailureInSendingToMQException {
        send(JSON.toJSONString(obj), topic, tag);
    }

    public void send(String body, String topic, String tag) throws FailureInSendingToMQException {
        send(body.getBytes(), topic, tag, false);
    }

    public void send(byte[] body, String topic, String tag, boolean delay) throws FailureInSendingToMQException {
        String key = UUID.randomUUID().toString();

        try {
            Message message = new Message(topic, tag, key, body);
            if (delay) {
                /*** 延迟5s ***/
                message.setDelayTimeLevel(2);
            }
            SendResult sendResult = producer.send(message);
            logger.info("已发送消息[topic={},tag={},key={},message={},发送状态为{}]", topic, tag, key, new String(body),
                    sendResult.getSendStatus());
            if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                throw new FailureInSendingToMQException(String.format("发送[key=%s,body=%s]到消息中间件失败,发送状态为%s",
                        key, body, sendResult.getSendStatus()));
            }
        } catch (InterruptedException | RemotingException | MQClientException | MQBrokerException e) {
            throw new FailureInSendingToMQException(
                    String.format("发送[key=%s,body=%s]到消息中间件失败", key, body), e);
        }
    }
}
